package com.jet.rtdb.elasticsearch.repository;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.jet.rtdb.elasticsearch.config.EsConfig;
import com.jet.rtdb.elasticsearch.eneity.Electric;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.net.UnknownHostException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.stream.Stream;

/**
 * @Author: zhangkaifeng.
 * @CreateTime: 2017/8/15 8:57
 * @Description:elasticsearchDAO
 */

@Repository
public class ESRepository {


    private static final Logger logger = LoggerFactory.getLogger(ESRepository.class);
    /**
     * jackson用于序列化操作的mapper
     */
    private static final ObjectMapper mapper = new ObjectMapper();

    /**
     * 查询
     */
    public List<Electric> search(String index, String type, Long stratTime, Long endTime) {
        Client client = null;
        List<Electric> electrics = new ArrayList<>();
        try {
            client = EsConfig.getInstance();
            QueryBuilder qb = new BoolQueryBuilder()
                    .must(QueryBuilders.rangeQuery("dateTime").gt(stratTime).lt(endTime));
            SearchResponse response = client.prepareSearch(index).setTypes(type).setQuery(qb).setSize(1000).execute()
                    .actionGet();
            SearchHit[] hits = response.getHits().getHits();
            for (SearchHit hit : hits) {
                Electric electric = mapper.readValue(hit.getSourceAsString(), Electric.class);
                electrics.add(electric);
            }
        } catch (UnknownHostException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return electrics;
    }

    /**
     * 新增document
     */
    public void addDocument(String index, String type, Electric electric) {
        Client client = null;
        try {
            client = EsConfig.getInstance();
            byte[] json = mapper.writeValueAsBytes(electric);
            client.prepareIndex(index, type, UUID.randomUUID().toString()).setSource(json).get();
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public void addDocument(String index, String type, List<Electric> electrics) {
        Client client = null;
        try {
            client = EsConfig.getInstance();
            for (Electric electric : electrics
                    ) {
                System.out.println("添加：" + electric);
                byte[] json = mapper.writeValueAsBytes(electric);
                client.prepareIndex(index, type, UUID.randomUUID().toString()).setSource(json).get();
            }
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 查询一个时间对应的数据
     *
     * @param index   数据库索引
     * @param type    索引的类型
     * @param time    查询时间
     * @param timeOut 超时时间（min）
     */
    public Electric queryByTime(String index, String type, Long time, int timeOut) {
        Client client = null;
        List<Electric> electrics = new ArrayList<>();
        Timestamp timestamp = new Timestamp(time);
        Electric electricResult = null;
        try {
            client = EsConfig.getInstance();
            SortBuilder sortBuilder = SortBuilders.fieldSort("dateTime")
                    .order(SortOrder.ASC);
            QueryBuilder qb = new BoolQueryBuilder()
                    .must(QueryBuilders.rangeQuery("dateTime").gt(time - timeOut * 60 * 1000).lt(time + timeOut * 60 * 1000));
            SearchResponse response = client.prepareSearch(index).setTypes(type).setQuery(qb).addSort(sortBuilder).setSize(1000).execute()
                    .actionGet();
            SearchHit[] hits = response.getHits().getHits();
            if (hits.length < 2) {
                return new Electric(Stream.of(index.toUpperCase(), ".", type).reduce("", String::concat), null, timestamp);
            }
            for (SearchHit hit : hits) {
                Electric electric = mapper.readValue(hit.getSourceAsString(), Electric.class);
                electrics.add(electric);
            }
            if (electrics.get(0).getDateTime().after(timestamp) || electrics.get(electrics.size() - 1).getDateTime().before(timestamp)) {
                return new Electric(Stream.of(index.toUpperCase(), ".", type).reduce("", String::concat), null, timestamp);
            }
            int start = 0;
            int end = electrics.size() - 1;
            for (int i = 0; i < electrics.size(); i++) {
                if (electrics.get(i).getDateTime().equals(timestamp)) {
                    return new Electric(Stream.of(index.toUpperCase(), ".", type).reduce("", String::concat), electrics.get(i).getValue(), timestamp);
                }
                if (electrics.get(i).getDateTime().after(timestamp)) {
                    start = i - 1;
                    break;
                }
            }
            for (int i = end; i > -1; i--) {
                if (electrics.get(i).getDateTime().equals(timestamp)) {
                    return new Electric(Stream.of(index.toUpperCase(), ".", type).reduce("", String::concat), electrics.get(i).getValue(), timestamp);
                }
                if (electrics.get(i).getDateTime().before(timestamp)) {
                    end = i + 1;
                    break;
                }
            }
            electricResult = getMatchData(electrics.get(start), electrics.get(end), time, timestamp);


        } catch (UnknownHostException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return electricResult;
    }

    /**
     * 查询最新值
     *
     * @param index   数据库索引
     * @param type    索引的类型
     */
    public Electric queryLastData(String index, String type) {
        Client client = null;
        List<Electric> electrics = new ArrayList<>();
        try {
            client = EsConfig.getInstance();
            SortBuilder sortBuilder = SortBuilders.fieldSort("dateTime")
                    .order(SortOrder.DESC);
            SearchResponse response = client.prepareSearch(index).setTypes(type).addSort(sortBuilder).setSize(1).execute()
                    .actionGet();
            SearchHit[] hits = response.getHits().getHits();
            if(hits.length == 0){
                return new Electric(Stream.of(index.toUpperCase(), ".", type).reduce("", String::concat), null, null);
            }
            for (SearchHit hit : hits) {
                Electric electric = mapper.readValue(hit.getSourceAsString(), Electric.class);
                electrics.add(electric);
            }
        } catch (UnknownHostException e) {
            e.printStackTrace();
        } catch (IndexNotFoundException e){
            logger.error("找不到{}索引", index);
            return new Electric(Stream.of(index.toUpperCase(), ".", type).reduce("", String::concat), null, null);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return electrics.get(0);
    }

    private Electric getMatchData(Electric electric1, Electric electric2, Long time ,Timestamp timestamp) {
        Long time1 = electric1.getDateTime().getTime();
        Long time2 = electric2.getDateTime().getTime();
        Double value1 = electric1.getValue();
        Double value2 = electric2.getValue();
        Double value = ((time-time1)*(value2 - value1))/(time2 - time1)+value1;
        return new Electric(electric1.getPointId(), value, timestamp);
    }
}
